package com.es.demo.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.annotation.TableName;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.baomidou.mybatisplus.extension.toolkit.SqlHelper;
import com.es.demo.entity.UserDO;
import com.es.demo.mapper.UserMapper;
import com.es.demo.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

/**
 * @author chenx
 * @date 2023/7/21-10:53
 */
@Service
@Slf4j
public class UserServiceImpl extends ServiceImpl<UserMapper, UserDO> implements UserService {
    @Resource
    private RestHighLevelClient restHighLevelClient;
    private static final int PAGE_LIMIT = 10000;

    @Override
    public UserDO getUserById(Integer id) {
        TableName table = UserDO.class.getAnnotation(TableName.class);
        String indexName = "";
        if (null != table) {
            indexName = table.value();
        } else {
            throw new RuntimeException("未知索引");
        }
        GetRequest getRequest = new GetRequest(indexName, String.valueOf(id));
        try {
            GetResponse response = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            String result = response.getSourceAsString();
            return JSON.parseObject(result, UserDO.class);
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        }
    }

    @Override
    public void userSqlDB2EsDB() {
        //查询所有的数据:如果
        int count = this.count();
        if (count > PAGE_LIMIT) {
            //TODO 数据太大需要进行分页
            System.err.println("数据太大需要进行分页");
            return;
        }
        List<UserDO> list = this.list();
        //约定表名即为es的索引名
        String tableName = SqlHelper.table(UserDO.class).getTableName();
        bulkWriteToElasticsearch(list, tableName);
    }

    public void bulkWriteToElasticsearch(List<UserDO> list, String indexName) {
        BulkRequest bulkRequest = new BulkRequest(indexName);
        try {
            for (UserDO user : list) {
                //使用hutool工具将实体转换为map,实体的id作为es的id
                IndexRequest indexRequest = new IndexRequest(indexName)
                        .source(BeanUtil.beanToMap(user))
                        .id(String.valueOf(user.getId()));
                bulkRequest.add(indexRequest);
            }
            BulkResponse response = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (response.hasFailures()) {
                log.error("write es error: {}", response.buildFailureMessage());
            }

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}